package com.tcs.mr.sumcount;

import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.tcs.mr.sumcount.map.TokenizerMapper;
import com.tcs.mr.sumcount.reducer.IntSumReducer;

public class WordCount {
	
	private static final Logger logger = LoggerFactory.getLogger(WordCount.class);
	
	public static void main(String [] args) throws IOException, ClassNotFoundException, InterruptedException {
		logger.info(" hadoop Application start , params : {} " , Arrays.toString(args));
		Configuration conf = new Configuration();
		String [] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();
		if (otherArgs.length < 2) {
			logger.error(" Usage : wordCount : 参数不能为空.请输入参数!" , new NullPointerException());
			System.exit(2);
		}
		@SuppressWarnings("deprecation")
		Job job = new Job(conf , "word count");
		job.setJarByClass(WordCount.class);
		job.setMapperClass(TokenizerMapper.class);
		job.setCombinerClass(IntSumReducer.class);
		job.setReducerClass(IntSumReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		for (int i = 0; i < otherArgs.length; i++) {
			FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
			FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
			System.exit(job.waitForCompletion(true) ? 0 : 1);
		}
	}
}